// -*-mode:c++; coding:utf-8-*-

#include "logger.hpp"

#include "row_mutator_on_tablet.hpp"

namespace xbase
{
	// 在 lower_level_service 上执行，除第一次异步提交外
	// （为优化，直接由上层调用，而不再转到自己的
	// lower_level_service 上调用）
	//
	// 否则，在 async_apply 中应该是这样调用 on_async_apply
	// this->get_lower_level_service().dispatch(boost::bind(&this_type::on_async_apply,
	// 						        this_ptr()));
	void row_mutator_on_tablet::on_async_apply()
	{
		m_session->async_write(m_req_ptr,
				       this->get_lower_level_service().wrap(
					       boost::bind(&this_type::on_response,
							   this_ptr(),
							   _1, _2)));
	}

	// 以下函数总在 lower_level_service 上执行

	void row_mutator_on_tablet::on_response(protocol::packet_ptr ptr, const error_code &socket_ec)
	{
		if(socket_ec) // 这是收发包出错，需要换session重试，由上层处理
		{
			on_error(socket_ec);
		}
		else if(ptr->get_type() != write_resp::type)
		{
			on_error(error::bad_packet_type);
		}
		else
		{
			const write_resp * resp = boost::polymorphic_downcast<write_resp*>(ptr.get());
			const protocol::write_response_v2 &body = resp->body();
			error_code ec = body.error_code();
			if(! ec)
			{
				dispatch(ec);
			}
			else if(ec == error::tablet_closed) // 需要重试
			{
				on_retry(ec);
			}
			else
			{
				on_error(ec);
			}
		}
	}

	void row_mutator_on_tablet::on_retry(const error_code &ec)
	{
		if(m_options.can_retry(m_retry_times))
		{
			LOG(WARN, "mutator fail: " << ec << " - " << ec.message()
			    << ", retry_times = " << m_retry_times);
			++ m_retry_times;
			if(! m_timer) // 延迟创建，因为大多数情况下不需要重试
				m_timer.reset(new boost::asio::deadline_timer(this->get_lower_level_service()));
			m_timer->expires_from_now(m_options.get_retry_interval());
			m_timer->async_wait(boost::bind(&this_type::on_async_apply,
							this_ptr()));
		}
		else
		{
			on_error(ec);
		}
	}

	void row_mutator_on_tablet::on_error(const error_code &ec)
	{
		LOG(ERROR, "mutator error: " << ec << " - " << ec.message()
		    << ", retry_times = " << m_retry_times
		    << ", request = "<< m_req_ptr->to_string());

		dispatch(ec);
	}

	void row_mutator_on_tablet::dispatch(const error_code &ec)
	{
		// release resource before dispatch
		m_session.reset();
		m_req_ptr.reset();
		// timer在下次提交时还会用到，暂不释放

		handler_type tmp = m_handler;
		m_handler = handler_type(); // clear handler
		this->get_upper_level_service().dispatch(
			boost::bind(tmp, ec)
			);
	}
}
